package com.haoziqi.chapter_06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.lang.reflect.MalformedParameterizedTypeException;
import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * description
 * created by A on 2021/3/5
 */
public class Flink06_App {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.读取数据
        DataStreamSource<MarketingUserBehavior> appDS = env.addSource(new AppData());
        /*
            1)需求:统计各个渠道各个行为
            2)数据维度:渠道、行为
                维度指的是看待问题的角度，维度字段对应的是度量值字段（事实）
         */
       appDS.map(new MapFunction<MarketingUserBehavior, Tuple2<String,Long>>() {
           @Override
           public Tuple2<String, Long> map(MarketingUserBehavior value) throws Exception {
               return Tuple2.of(value.getChannel()+"_"+value.getBehavior(),1L);
           }
       }).keyBy(data -> data.f0).sum(1).print();

        env.execute();
    }

    public static class AppData implements SourceFunction<MarketingUserBehavior> {

        private static volatile boolean isRunning = true;
        private List<String> userBehaviorList = Arrays.asList("DOWNLOAD", "INSTALL", "UNINSTALL", "UPDATE");
        private List<String> channelList = Arrays.asList("XIAOMI", "HUAWEI", "APPSTORE", "OPPO", "VIVO", "ZHUSHOU");

        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            Random random = new Random();
            while (isRunning) {
                ctx.collect(
                        new MarketingUserBehavior(
                                random.nextLong(),
                                userBehaviorList.get(random.nextInt(userBehaviorList.size())),
                                channelList.get(random.nextInt(channelList.size())),
                                System.currentTimeMillis())
                );
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
